/*
 Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
 Original sources are available at www.latestbit.com

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

 TCPConnection.cpp. Created on: 20.08.2009 15:30:36
 */

#include "smartobjects/network/tcp/TCPConnection.hpp"

namespace smartobjects {

const int TCPConnection::READ_BUFFER_SIZE 								= 65535;
const int TCPConnection::WRITE_BUFFER_SIZE 								= 65535;
const char* TCPConnection::CONNECTION_MULTIPLEXER_GROUP_NAME 			= "CONNECTIONS";
const char* TCPConnection::CONNECTION_SENDER_THREAD_POOL_GROUP_NAME 	= "CONNECTION_WRITERS";
const int TCPConnection::KEEP_ALIVE_MAX_INACTIVITY_TIMEOUT				= 15; // seconds

TCPConnection::TCPConnection( TCPSharedResources::SharedPtr poSharedResources ) :
			m_poSharedResources ( poSharedResources ) ,
			m_hSocket(-1),
			m_poSocketURI(NULL),
			m_bStoppedState(false),
			m_bIsMultiplexed(false),
			m_poPacketFormer(NULL),
			m_poHandshakeManager(NULL),
			m_poDataEncrypter (NULL),
			m_poUserCredentials (NULL) {

	m_poReadBuffer = new ByteBuffer( READ_BUFFER_SIZE );
	m_poWriteBuffer = new ByteBuffer( WRITE_BUFFER_SIZE );

	m_poThreadPool = m_poSharedResources->getThreadPool ( CONNECTION_SENDER_THREAD_POOL_GROUP_NAME );
	m_nSenderTaskId = m_poThreadPool->createTaskId();
	m_poMultiplexer = m_poSharedResources->getSocketMultiplexer( CONNECTION_MULTIPLEXER_GROUP_NAME );

	if(m_poSharedResources->getPacketFormerFactory()) {
		m_poPacketFormer = m_poSharedResources->getPacketFormerFactory()->createPacketFormer();
	}
	if(m_poSharedResources->getHandshakeManagerFactory()) {
		m_poHandshakeManager = m_poSharedResources->getHandshakeManagerFactory()->createHandshakeManager(this);
	}

	m_nLastActivityTime = time(NULL);
}

TCPConnection::~TCPConnection() {
	// close();
	releaseAndClear(m_poSocketURI);
	releaseAndClear(m_poDataEncrypter);
	releaseAndClear(m_poUserCredentials);

	releaseAndClear(m_poReadBuffer);
	releaseAndClear(m_poWriteBuffer);

	if(m_poPacketFormer) {
		delete m_poPacketFormer;
		m_poPacketFormer = NULL;
	}

	if( m_poHandshakeManager ) {
		delete m_poHandshakeManager;
		m_poHandshakeManager = NULL;
	}

	m_poMultiplexer.reset();
	m_poThreadPool.reset();
	m_poSharedResources.reset();
}

void TCPConnection::addConnectionListener(IConnectionListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxConnectionListeners );
	if( !m_oPALConnectionListeners.isActivated() ) {
		m_aConnectionListeners.push_back(poListener);
	}
	else {
		m_oPALConnectionListeners.putListenerToAdd(poListener);
	}
}

void TCPConnection::delConnectionListener(IConnectionListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxConnectionListeners );
	if( !m_oPALConnectionListeners.isActivated() ) {
		m_aConnectionListeners.remove(poListener);
	}
	else {
		m_oPALConnectionListeners.putListenerToRemove(poListener);
	}

}

void TCPConnection::addStreamListener(IStreamListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxStreamListeners );
	if( !m_oPALStreamListeners.isActivated() ) {
		m_aStreamListeners.push_back(poListener);
	}
	else {
		m_oPALStreamListeners.putListenerToAdd(poListener);
	}
}

void TCPConnection::delStreamListener(IStreamListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxStreamListeners );
	if( !m_oPALStreamListeners.isActivated() ) {
		m_aStreamListeners.remove(poListener);
	}
	else {
		m_oPALStreamListeners.putListenerToRemove(poListener);
	}
}

IHandshakeManager* TCPConnection::getHandshakeManager() {
	return m_poHandshakeManager;
}

IPacketFormer* TCPConnection::getPacketFormer() {
	return m_poPacketFormer;
}

IDataEncrypter* TCPConnection::getDataEncrypter() {
	return m_poDataEncrypter;
}

void TCPConnection::setDataEncrypter(IDataEncrypter* poDataEncrypter) {
	setNewRefObject(m_poDataEncrypter, poDataEncrypter);
}

IUserCredentials* TCPConnection::getUserCredentials() {
	return m_poUserCredentials;
}

void TCPConnection::setUserCredentials(IUserCredentials* poUserCredentials) {
	setNewRefObject(m_poUserCredentials, poUserCredentials);
}

void TCPConnection::setEndpointURI(smartobjects::sockets::SocketURI* value) {
	setNewRefObject(m_poSocketURI, value);
}

URI* TCPConnection::getEndpointURI(void) {
	return m_poSocketURI;
}

void TCPConnection::setSocket(smartobjects::sockets::Socket hSocket) {
	m_hSocket = hSocket;
}

smartobjects::sockets::Socket TCPConnection::getSocket() {
	return m_hSocket;
}

void TCPConnection::start() throw (const IOException&)  {
	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );
	m_bStoppedState = false;
	oAsyncEventLock.unlock();

	boost::recursive_mutex::scoped_lock oLock(m_mxSocketLock);

	smartobjects::sockets::osimpl::setNoneBlockMode ( m_hSocket, true );

	if(!isOpen()) {
		oLock.unlock();
		boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );
		m_bStoppedState = true;
		throw IOException("Unable to start closed connection");
	}

	m_nLastActivityTime = time(NULL);
	m_poMultiplexer->addSocketReader ( m_hSocket,
			boost::bind ( &TCPConnection::onAsyncReceive, this, _1 ),
			boost::bind ( &TCPConnection::onCheckKeepAlive, this )
	);
	m_bIsMultiplexed = true;
}

void TCPConnection::close(void) {
	bool hasSomeDataToWrite = false;
	boost::recursive_mutex::scoped_lock oWriteBufferLock(m_mxWriteBufferLock);
	hasSomeDataToWrite =  m_poWriteBuffer->getLen() > 0;
	oWriteBufferLock.unlock();

	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );

	m_bStoppedState = true;
	oAsyncEventLock.unlock();
	boost::recursive_mutex::scoped_lock oLock(m_mxSocketLock);
	if(this->isOpen()) {
		if(m_bIsMultiplexed) {
			m_poMultiplexer->removeSocketReader ( m_hSocket );
			m_bIsMultiplexed = false;
		}
		if(hasSomeDataToWrite)
			boost::this_thread::sleep(boost::posix_time::seconds(3));
		smartobjects::sockets::osimpl::shutdown(m_hSocket);
		smartobjects::sockets::osimpl::close(m_hSocket);
		m_hSocket = -1;
	}

	if( m_poHandshakeManager ) {
		m_poHandshakeManager->reset();
	}

	oLock.unlock();

	m_poThreadPool->cancel ( m_nSenderTaskId );
	m_poThreadPool->wait (m_nSenderTaskId );

	oWriteBufferLock.lock();
	m_poWriteBuffer->clear();
}

bool TCPConnection::isOpen(void) {
	boost::recursive_mutex::scoped_lock oLock(m_mxSocketLock);
	return m_hSocket!=-1;
}

void TCPConnection::onAsyncReceive(bool bIsOOB) throw() {

	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );

	if ( m_bStoppedState )
		return;
	m_nLastActivityTime = time(NULL);

	SafeForReleaseMethodMutexLock< boost::recursive_mutex > oReleaseLock(this , oAsyncEventLock );
	int nReadedBytes;
	do {
		nReadedBytes = 0;
		boost::recursive_mutex::scoped_lock oLock(m_mxSocketLock);
		if(isOpen() && !m_bStoppedState) {
			int nOOBReadedBytes = 0;
			if(bIsOOB ) {
				nOOBReadedBytes = smartobjects::sockets::osimpl::recv (
						m_hSocket,
						( m_poReadBuffer->getData() + m_poReadBuffer->getLen() ),
						m_poReadBuffer->getAvailableCapacity(),
						MSG_OOB
				);
			}

			nReadedBytes = smartobjects::sockets::osimpl::recv (
					m_hSocket,
					( m_poReadBuffer->getData() + m_poReadBuffer->getLen() ),
					m_poReadBuffer->getAvailableCapacity()
			);

			if(bIsOOB &&  nReadedBytes < 0 && nOOBReadedBytes > 0) {
				return;
			}
		}
		oLock.unlock();

		if (nReadedBytes > 0) {
			m_poReadBuffer->setLen ( nReadedBytes );

			if( !m_poPacketFormer ) {
				fireOnReceived(m_poReadBuffer);
			}
			else {
				m_poPacketFormer->onReceiveRawPacket(m_poReadBuffer);
				ByteBuffer* poFormedBuffer = NULL;
				try {
					do {
						poFormedBuffer = m_poPacketFormer->getReadyPacket();
						if(poFormedBuffer!=NULL) {
							fireOnReceived(poFormedBuffer);
						};
					}
					while(poFormedBuffer!=NULL);
				}
				catch(const InvalidPacketReceivedException& ex) {
					fireOnReceiveError( ex.what() );
					fireOnDisconnected();
					close();
				}
			}
			m_poReadBuffer->setLen(0);
		}
		else {
			if(nReadedBytes < 0) {
				fireOnReceiveError( getLastOSErrorString() );
			}
			fireOnDisconnected();
			close();
		}
	}
	while( nReadedBytes >= (int)m_poReadBuffer->getAvailableCapacity() );
}


void TCPConnection::send(const ByteBuffer* poBuffer) throw (const IOException&)  {
	if(!isOpen())
		throw IOException("Unable to send packet for closed connection!");

	if( poBuffer->getLenFromPos() > 0 ) {

		boost::recursive_mutex::scoped_lock oLock(m_mxWriteBufferLock);
		bool bIsNeededScheduling = m_poWriteBuffer->getLenFromPos() <= 0;
		m_poWriteBuffer->append ( poBuffer->getDataAtPos(), poBuffer->getLenFromPos() );

		oLock.unlock();

		if (bIsNeededScheduling) {
			m_poThreadPool->schedule ( m_nSenderTaskId,
				boost::bind ( &TCPConnection::onAsyncWrite, this )
			);
		}
	}
}

void TCPConnection::onAsyncWrite() throw() {
	int nBytesSent = 0;
	bool bTrying = false;
	// int nRetryDelay = -10;

	try {
		boost::recursive_mutex::scoped_lock oLock(m_mxWriteBufferLock);
		if(m_bStoppedState)
			return;
		do {
			bTrying = false;

			nBytesSent = 0;
			if( m_poWriteBuffer->getLenFromPos() > 0 ) {
				boost::recursive_mutex::scoped_lock oSocketLock(m_mxSocketLock);
				if(isOpen())
					nBytesSent = smartobjects::sockets::osimpl::send ( m_hSocket, m_poWriteBuffer->getDataAtPos(), m_poWriteBuffer->getLenFromPos());
				else
					throw IOException("Unable to write data to closed socket");
			}

			if ( nBytesSent >= 0) {
				m_poWriteBuffer->seek ( nBytesSent );
				if ( m_poWriteBuffer->getLenFromPos() > 0 ) {
					/*oLock.unlock();
					if( nRetryDelay > 0) {
						//xtSleepMsec(nRetryDelay);
					}
					if(nRetryDelay < 50)
						nRetryDelay+=10;
					bTrying = true;
					oLock.lock();*/
					bTrying = true;
				}
			}
			else
			if (nBytesSent == -1) {
				throw IOException("Unable to write data to socket: "+getLastOSErrorString());
			}
		}
		while(bTrying);
		m_poWriteBuffer->clear();
	}
	catch(const IOException& ex) {
		fireOnWriteError(ex.what());
	}
}

void TCPConnection::onCheckKeepAlive() throw() {
	boost::recursive_mutex::scoped_lock oAsyncEventLock ( m_mxAsyncEvent );
	if ( m_bStoppedState )
		return;
	time_t now = time(NULL);
	if ( m_nLastActivityTime + KEEP_ALIVE_MAX_INACTIVITY_TIMEOUT < now  ) {
		boost::recursive_mutex::scoped_lock oSocketLock(m_mxSocketLock);
		if(isOpen()) {
			unsigned char nKeepAliveData = 0xAA;
			smartobjects::sockets::osimpl::send (
					m_hSocket,
					&nKeepAliveData,
					1,
					MSG_OOB
			);
		}
		m_nLastActivityTime = now;
	}
}


bool TCPConnection::isHandshaked() {
	return m_poHandshakeManager==NULL || m_poHandshakeManager->isHandshaked();
}

void TCPConnection::fireOnDisconnected() {
	if(isHandshaked()) {
		boost::recursive_mutex::scoped_lock oLock(m_mxConnectionListeners);
		m_oPALConnectionListeners.activate();
		foreach(IConnectionListener* poListener, m_aConnectionListeners) {
			poListener->onDisconnected(this);
		}
		processingPostActionConnectionListeners();
	}
}

void TCPConnection::fireOnReceiveError(const std::string& sErrorMessage) {
	if(isHandshaked()) {
		boost::recursive_mutex::scoped_lock oLock(m_mxStreamListeners);
		m_oPALStreamListeners.activate();
		foreach(IStreamListener* poListener, m_aStreamListeners) {
			poListener->onReceiveError(this, sErrorMessage);
		}
		processingPostActionStreamListeners();
	}
}

void TCPConnection::fireOnWriteError(const std::string& sErrorMessage) {
	if(isHandshaked()) {
		boost::recursive_mutex::scoped_lock oLock(m_mxStreamListeners);
		m_oPALStreamListeners.activate();
		foreach(IStreamListener* poListener, m_aStreamListeners) {
					poListener->onWriteError(this, sErrorMessage);
		}
		processingPostActionStreamListeners();
	}
}

void TCPConnection::fireOnReceived( ByteBuffer* poBuffer ) {
	if(m_poHandshakeManager!=NULL && !m_poHandshakeManager->isHandshaked()) {
		try {
			m_poHandshakeManager->onHandshakeReceived ( poBuffer );
			if(m_poHandshakeManager->isHandshaked()) {
				fireOnConnected();
			}
		}
		catch(const HandshakeException&) {
			close();
		}
	}
	else
	if(isHandshaked()) {
		boost::recursive_mutex::scoped_lock oLock(m_mxStreamListeners);
		m_oPALStreamListeners.activate();
		foreach(IStreamListener* poListener, m_aStreamListeners) {
			poBuffer->mark();
			poListener->onReceived(this, poBuffer);
			poBuffer->resetToMarker();
		}
		processingPostActionStreamListeners();
	}
}

void TCPConnection::processingPostActionConnectionListeners() {
	m_oPALConnectionListeners.deactivate();
	foreach( PostActionListenersContainer<IConnectionListener>::PostActionQueueItem item, m_oPALConnectionListeners.getPostActionQueue()) {
		if(item.isActionToAdd()) {
			addConnectionListener( item.getListener() );
		}
		else
		if(item.isActionToRemove()) {
			delConnectionListener( item.getListener() );
		}
	}
	m_oPALConnectionListeners.reset();
}

void TCPConnection::processingPostActionStreamListeners() {
	m_oPALStreamListeners.deactivate();
	foreach( PostActionListenersContainer<IStreamListener>::PostActionQueueItem item, m_oPALStreamListeners.getPostActionQueue()) {
		if(item.isActionToAdd()) {
			addStreamListener( item.getListener() );
		}
		else
		if(item.isActionToRemove()) {
			delStreamListener( item.getListener() );
		}
	}
	m_oPALStreamListeners.reset();
}

}
